package cn.getech.data.metadata.db;

import cn.getech.data.development.entity.ConfConnect;
import cn.getech.data.development.utils.HdfsUtil;
import cn.getech.data.intelligence.common.exception.RRException;
import cn.getech.data.metadata.constant.ConnectTypeEnum;
import cn.getech.data.metadata.entity.meta.CollectionTableField;
import cn.getech.data.metadata.entity.meta.CollectionTableInfo;
import cn.getech.data.metadata.entity.meta.CollectionTask;
import cn.getech.data.metadata.utils.SingleConnect;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;

import java.math.BigDecimal;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.List;
@Slf4j
public class HIveStructure extends DBStructure {
    private ConfConnect confConnect;

    public HIveStructure(ConfConnect confConnect, String url, CollectionTask collectionTask) {
        super(url, confConnect.getUsername(), confConnect.getPassword(),
                ConnectTypeEnum.Hive.getDriverClass(), confConnect.getDbname(), collectionTask);
        this.confConnect = confConnect;
    }
    @Override
    protected CollectionTableInfo getRowsAndLength(CollectionTableInfo tableInfo) {
        String dirHome = "/user/hive/warehouse/";
        HdfsUtil hdfsUtil = null;
        BigDecimal tableSize=new BigDecimal(0);
        try {
            if (confConnect != null) {
                hdfsUtil = new HdfsUtil(confConnect.getNamespace(), confConnect.getNameNode(),confConnect.getNameNodePort());
                String dbUrl = dirHome + tableInfo.getDbName() + ".db";
                Long tableSizeLong = hdfsUtil.getTableOrDbSize(dbUrl + "/" + tableInfo.getTableName());
                if(tableSizeLong!=null){
                    tableSize=new BigDecimal(tableSizeLong);
                }
                tableInfo.setTotalSize(tableSize.doubleValue());
            }
        } catch (Exception e) {
          log.warn("{}统计大小失败:{}",tableInfo.getTableName(),e);
        }finally {
            if(hdfsUtil != null){
                hdfsUtil.close();
            }
        }
        return tableInfo;
    }

    @Override
    protected List<CollectionTableInfo> getSourceTables(String db) throws Exception {
        List<CollectionTableInfo> list = new ArrayList<>();
        Connection conn = null;
        try {

            conn = SingleConnect.getInstance().getConnection(url, username, password, driverClass);
            if (conn == null)
                throw new RRException("获取数据库连接失败");
            ResultSet result;
            Statement st = conn.createStatement();
            st.execute("use "+db);
            result = st.executeQuery("show tables");
            while (result != null && result.next()) {
                CollectionTableInfo tableInfo = new CollectionTableInfo();
                tableInfo.setDbName(db);
                tableInfo.setTableType(3);
                tableInfo.setTableName(result.getString("tab_name"));
                String remarks = result.getString("tab_name");
                tableInfo.setTableDesc(remarks);
                if (StringUtils.isEmpty(remarks))
                    tableInfo.setTableAlias(tableInfo.getTableName());
                else
                    tableInfo.setTableAlias(remarks);
                list.add(tableInfo);
            }
        } catch (Exception e) {
            throw e;
        } finally {
            try {
                if (conn != null) {
                    conn.close();
                }
            } catch (SQLException e) {
                log.error(e.getMessage());
            }
        }
        return list;
    }

    @Override
    protected List<CollectionTableField> getSourceColumns(CollectionTableInfo tableInfo) {
        List<CollectionTableField> list = new ArrayList<>();
        Connection conn = null;
        try {
            conn = SingleConnect.getInstance().getConnection(url, username, password, driverClass);
            if (conn == null)
                throw new RRException("获取数据库连接失败");
            ResultSet result;
            Statement st = conn.createStatement();
            result = st.executeQuery("desc " + tableInfo.getDbName() + "." + tableInfo.getTableName());
            while (result != null && result.next()) {
                String columnName = result.getString("col_name");
                if (StringUtils.isEmpty(columnName)) {
                    continue;
                }
                if (columnName.startsWith("#")) {
                    continue;
                }
                String typeName = result.getString("data_type");
                if (StringUtils.isEmpty(typeName)) {
                    continue;
                }
                CollectionTableField tableField = new CollectionTableField();
                tableField.setFieldName(columnName);
                tableField.setFieldType(typeName);
                tableField.setFieldDesc(result.getString("comment"));
                //外部采集的数据，肯定没有分区字段
                tableField.setIsPartition(0);
                //设置是否为空
                tableField.setFieldIsNull(0);
                tableField.setFieldLength(0);
                //设置是否主键
                tableField.setPrimaryKey(0);
                list.add(tableField);
            }
        } catch (Exception e) {
            log.error("", e);
        } finally {
            try {
                if (conn != null) {
                    conn.close();
                }
            } catch (SQLException e) {
                log.error("", e);
            }
        }
        //设置默认数据密级
        return getCollectionTableFields(tableInfo, list);
    }

}
